package 大数据比赛样题_2020

import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.expressions.Window

object test03 {
  def main(args: Array[String]): Unit = {
    // 创建配置环境
    val spark =SparkSession.builder()
      .appName("score_count")
      .master("local[6]")
      .getOrCreate()
    import spark.implicits._

    // 读取数据集
    // 分数信息
    val score = spark.sparkContext.textFile("hdfs://192.168.64.129:9000/user/root/testdata/score.txt")
    val data1 = score.map(item=>((item.split(",")(0)),(item.split(",")(1)),(item.split(",")(2).toDouble)) )
      .toDF("学号","科目","成绩")
    data1.show()
    // 学生信息
    val student = spark.sparkContext.textFile("hdfs://192.168.64.129:9000/user/root/testdata/students.txt")
    val data2 = student.map(item=>((item.split(",")(0)),(item.split(",")(1)),(item.split(",")(4))) )
      .toDF("学号","姓名","班级")
    data2.show()

    // 合并分数、学生信息
    val data_split = data1.join(data2,data1.col("学号") === data2.col("学号"))
      .select(data1.col("学号"),data1.col("科目"),data1.col("成绩"),data2.col("姓名"),data2.col("班级"))
        .where('成绩 > 60.0)
    data_split.show()

    import org.apache.spark.sql.functions._
    // 使用窗口函数进行分区统计
    val window = Window.partitionBy('学号)
    val data60 = data_split.select('学号,'姓名,'班级,'科目,'成绩,count('成绩) over(window) as "count")
    data60.show()


    // 查找出6门科目均合格的学生姓名
    val pass_student = data60.select('学号,'姓名,'班级,'科目,'成绩)
      .where('count === 6)
      .show()


    //pass_student.write.csv("src/main/scala/大数据比赛样题/passAll1_count")


//    // 合并分数、学生信息
//    val data_split = data1.join(data2,data1.col("学号") === data2.col("学号"))
//      .select(data1.col("学号"),data1.col("科目"),data1.col("成绩"),data2.col("姓名"),data2.col("班级"))
//    data_split.show()
//
//    import org.apache.spark.sql.functions._
//    // 查找各科成绩凡是大于60的
//    val data60 = data_split.select('学号,'姓名,'班级,'科目,'成绩)
//    .where('成绩 > 60.0)
//      .groupBy('学号,'姓名,'班级)
//      .count().as('count)
//    data60.show()
//
//    // 查找出6门科目均合格的学生姓名
//    val pass_student = data60.select('姓名)
//    .where('count === 6)
//    pass_student.show()
//
//    // 通过合格学生姓名，合并信息
//    val final_pass = data_split.join(pass_student,data_split.col("姓名") === pass_student.col("姓名"))
//      .select(data_split.col("学号"),data_split.col("姓名"),data_split.col("班级"),data_split.col("科目"),data_split.col("成绩"))
//    final_pass.show()
//
//    final_pass.write.csv("src/main/scala/大数据比赛样题/passAll_count")
  }
}
